package com.etlmaster.executor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.etlmaster.executor.steps.control.ConditionalStepInterface;
import com.etlmaster.executor.steps.control.LoopStepInterface;
import com.etlmaster.executor.utils.CallableResult;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.executor.utils.MultiProcessHelper;
import com.etlmaster.executor.utils.StackEntity;
import com.etlmaster.executor.utils.StepState;
import com.etlmaster.executor.utils.ThreadEntity;

/**
 * @author Larry
 * @date 2017-04-14 步骤并发核心类
 */
public class StepStack {
	private Lock lock = new ReentrantLock();// 锁对象
	private Condition notEmpty = getLock().newCondition();

	private volatile boolean isAllFetched=false;
	private String id;

	private boolean isLoopStack;

	public static Map<String,StepStack> stepStacks=new HashMap<String,StepStack>();
	private List<CallableResult> result = new ArrayList<CallableResult>();
	private MultiProcessHelper helper;

	private StackEntity stackThread;
	public StepStack(String id, int size, boolean isloopStack) {
		this.setId(id + UUID.randomUUID().toString());
		this.setLoopStack(isloopStack);
		this.setHelper(new MultiProcessHelper(getSize(size)));
		//start();
		this.setStackThread(new StackEntity(this));
		stepStacks.put(id, this);
	}
	
	public class MetaEntry {// 步骤栈中存储的对象
		private StepMeta meta;
		private Job job;
		private String id;

		private Future<CallableResult> future = null;

		private MetaEntry(StepMeta meta, Job job, String id) {
			this.meta = meta;
			this.job = job;
			this.setId(id);
		}

		public StepMeta getMeta() {
			return meta;
		}
		public void setMeta(StepMeta meta) {
			this.meta=meta;
		}
		public Job getJob() {
			return job;
		}

		public Future<CallableResult> getFuture() {
			return future;
		}

		public void setFuture(Future<CallableResult> future) {
			this.future = future;
		}

		public String getId() {
			return id;
		}

		public void setId(String id) {
			this.id = id;
		}
	}

	private boolean containsEntry(String metaid) {
		boolean isExist = false;
		for (Object entry : this.getHelper().getWaitQueue()) {
			MetaEntry meta = (MetaEntry) entry;
			if (meta.getId().equals(metaid)) {
				isExist = true;
				break;
			}
		}
		return isExist;
	}

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public void pushStack(BlockingQueue stack, StepMeta meta, Job job, String id) {
		if (isLoopStack()) {
			MetaEntry entry = new MetaEntry(meta, job, id);
			stack.add(entry);

		} else if (!isLoopStack() && !containsEntry(meta.getId())) {
			MetaEntry entry = new MetaEntry(meta, job, id);
			stack.add(entry);

		} else {
			LogWriter.addLog(job.getMeta().getId(),0,"WARN", "{}不满足运行情况！", meta.getId());
		}
	}

	public void pushStack(BlockingQueue stack, List<StepMeta> metas, Job job,String sub_session_id) {
		for (StepMeta meta : metas) {
			if (DependencyCheck.isPreDependentReady(sub_session_id,meta, job))
				pushStack(stack, meta, job, sub_session_id);
			else {
				LogWriter.addLog(job.getMeta().getId(),0,"WARN", "前置没准备好.{}", meta.getId());
			}
		}
	}

	// 检查栈里的步骤的运行情况，获取下一运行的栈信息
	public boolean handleStack(MetaEntry entry) throws InterruptedException {

		StepState stepState = entry.meta.getStepState();

		// if (stepState.isDone()) {
		if (stepState.getReturnValue() != null
				&& stepState.getReturnValue().size() > 0) {// 把步骤的返回值加入环境的变量
			JobPool.getVariable(entry.job.getSeesionId()).putAll(
					stepState.getReturnValue());
		}
		if (entry.meta instanceof ConditionalStepInterface) {// 如果是条件节点，需要把下一节点置成当前节点
			entry.meta = ((ConditionalStepInterface) entry.meta)
					.processForStack(entry.meta, entry.job);
			stepState = entry.meta.getStepState();
			if (stepState.isDone()) {
				if (stepState.getReturnValue() != null
						&& stepState.getReturnValue().size() > 0) {// 把步骤的返回值加入环境的变量
					JobPool.getVariable(entry.job.getSeesionId()).putAll(
							stepState.getReturnValue());
				}
			}
		}
		List<StepMeta> nextSteps = null;
		if (entry.meta instanceof LoopStepInterface) {
			nextSteps = ((LoopStepInterface) entry.meta)
					.getEndLoopNextSteps(entry.job);
		} else {
			nextSteps = entry.job.getNextExecStepMetas(entry.meta.getId(),
					stepState);
		}
		if (nextSteps != null && nextSteps.size() > 0) {

			pushStack(getHelper().getWaitQueue(), nextSteps, entry.job,entry.getId());

		} 
		else if(this.isLoopStack&&entry.job.isTmpJob()){
			JobPool.copyVariables(entry.job.getSeesionId(), entry.getId().split("_")[0]);
			JobPool.destoryVariable(entry.job.getSeesionId());
		}
		return false;

	}

	public void cosumeQueue() {
		getLock().lock();
		while (!getHelper().getWaitQueue().isEmpty()) {
			try {
				if(getHelper().getRunningQueue().remainingCapacity()>0)
				{
					MetaEntry entry = (MetaEntry) getHelper().getWaitQueue().poll();
					ThreadEntity thread = new ThreadEntity(this, entry, getHelper());

					getHelper().getRunningQueue().put(entry);
				
					getHelper().submitTask(thread, entry);
				}else{
					getNotEmpty().await(1000,TimeUnit.MILLISECONDS);
					getLock().lock();
					if (this.getResult().size() > 0) {
						checkResult();
					} else {
						getLock().unlock();
					}
				}
			} catch (InterruptedException e1) {
				LogWriter.addLog("ERROR", e1.getMessage());
			}

		}

		if (!getHelper().getRunningQueue().isEmpty()) {
			try {
				getNotEmpty().await(1000,TimeUnit.MILLISECONDS);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		if (this.getResult().size() > 0) {
			checkResult();
		} else {
			getLock().unlock();
		}

	}

	private void checkResult() {
		for (CallableResult rst : result) {
			if (rst.isError()) {
				getLock().unlock();

				throw new EtlMasterException(rst.getMessage());
			} else {
				try {
					handleStack(rst.getEntry());
					getHelper().getRunningQueue().poll();
				} catch (Exception e) {
					getLock().unlock();
					throw new EtlMasterException(e.getMessage());
				}
			}
		}
		result.clear();
		getLock().unlock();

	}

	public void startStack() {
		Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
			public void uncaughtException(Thread t, Throwable e) {
				LogWriter.addLog("ERROR", "{},{}", t.getName(), e.getMessage());
			}
		});
		try {
			while (!helper.getRunningQueue().isEmpty()
					|| !helper.getWaitQueue().isEmpty()||!isAllFetched) {
				cosumeQueue();

			}
		} catch (Exception e) {
			e.printStackTrace();
			throw new EtlMasterException(e.getMessage());
		}
	}

	public int getSize(int size) {
		if (size > DataCollectionUtils.getIntProperties("maxStackSize")
				|| size < 1) {
			return DataCollectionUtils.getIntProperties("defaultStackSize");
		} else {
			return size;
		}
	}

	public boolean isLoopStack() {
		return isLoopStack;
	}

	public void setLoopStack(boolean isLoopStack) {
		this.isLoopStack = isLoopStack;
	}

	public MultiProcessHelper getHelper() {
		return helper;
	}

	public void setHelper(MultiProcessHelper helper) {
		this.helper = helper;
	}

	public Lock getLock() {
		return lock;
	}

	public void setLock(Lock lock) {
		this.lock = lock;
	}

	public Condition getNotEmpty() {
		return notEmpty;
	}

	public void setNotEmpty(Condition notEmpty) {
		this.notEmpty = notEmpty;
	}


	public List<CallableResult> getResult() {
		return result;
	}

	public void setResult(List<CallableResult> result) {
		this.result = result;
	}

	public StackEntity getStackThread() {
		return stackThread;
	}
	public void setStackThread(StackEntity stackThread) {
		this.stackThread = stackThread;
	}

	public boolean isAllFetched() {
		return isAllFetched;
	}

	public void setAllFetched(boolean isAllFetched) {
		this.isAllFetched = isAllFetched;
	}

}
